// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

/**
 * @file HelloWorldSubscriber.cpp
 *
 */
#include "MessagePubSubTypes.h"
#include "MessageSubscriber.h"
//#include <fastrtps/participant/Participant.h>
//#include <fastrtps/attributes/ParticipantAttributes.h>
//#include <fastrtps/attributes/SubscriberAttributes.h>
//#include <fastrtps/subscriber/Subscriber.h>
//#include <fastrtps/Domain.h>

#include "logger.h"

HelloWorldSubscriber::HelloWorldSubscriber()
        : m_participant(nullptr)
        , m_subscriber(nullptr)
        , m_sampleType(new HelloWorldPubSubType())
{
}

bool HelloWorldSubscriber::init()
{
#if 0
    ParticipantAttributes PParam;
    PParam.rtps.builtin.discovery_config.discoveryProtocol = DiscoveryProtocol_t::SIMPLE;
    PParam.rtps.builtin.discovery_config.use_SIMPLE_EndpointDiscoveryProtocol = true;
    PParam.rtps.builtin.discovery_config.m_simpleEDP.use_PublicationReaderANDSubscriptionWriter = true;
    PParam.rtps.builtin.discovery_config.m_simpleEDP.use_PublicationWriterANDSubscriptionReader = true;
    PParam.rtps.builtin.discovery_config.leaseDuration = c_TimeInfinite;
    PParam.rtps.setName("Participant_sub");
    mp_participant = Domain::createParticipant(PParam);
    if (mp_participant == nullptr)
    {
        return false;
    }

    //REGISTER THE TYPE

    Domain::registerType(mp_participant, &m_type);
    //CREATE THE SUBSCRIBER
    SubscriberAttributes Rparam;
    Rparam.topic.topicKind = NO_KEY;
    Rparam.topic.topicDataType = "HelloWorld";
    Rparam.topic.topicName = "HelloWorldTopic";
    Rparam.topic.historyQos.kind = KEEP_LAST_HISTORY_QOS;
    Rparam.topic.historyQos.depth = 30;
    Rparam.topic.resourceLimitsQos.max_samples = 50;
    Rparam.topic.resourceLimitsQos.allocated_samples = 20;
    Rparam.qos.m_reliability.kind = RELIABLE_RELIABILITY_QOS;
    Rparam.qos.m_durability.kind = TRANSIENT_LOCAL_DURABILITY_QOS;
//    mp_subscriber = Domain::createSubscriber(mp_participant, Rparam, (SubscriberListener*)&m_listener);
    mp_subscriber = Domain::createSubscriber(mp_participant, Rparam, nullptr);

    if (mp_subscriber == nullptr)
    {
        return false;
    }
#else
    DomainParticipantQos psqos;
    psqos.wire_protocol().builtin.discovery_config.discoveryProtocol = DiscoveryProtocol_t::SIMPLE;
    psqos.wire_protocol().builtin.discovery_config.use_SIMPLE_EndpointDiscoveryProtocol = true;
    psqos.wire_protocol().builtin.discovery_config.m_simpleEDP.use_PublicationReaderANDSubscriptionWriter = true;
    psqos.wire_protocol().builtin.discovery_config.m_simpleEDP.use_PublicationWriterANDSubscriptionReader = true;
    psqos.wire_protocol().builtin.discovery_config.leaseDuration = eprosima::fastrtps::c_TimeInfinite;
//    psqos.wire_protocol().builtin.discovery_config.leaseDuration = eprosima::fastrtps::c_TimeZero;

    psqos.name("SubscriberParticipant");

    m_participant = DomainParticipantFactory::get_instance()->create_participant(0, psqos);
    if (m_participant == nullptr)
    {
        Logger_Error("Something went wrong while creating the Subscriber Participant...");
        return false;
    }

    //Register the Type
    m_sampleType.register_type(m_participant);

    //Keep All Sub
    m_subscriber = m_participant->create_subscriber(SUBSCRIBER_QOS_DEFAULT);
    if (m_subscriber == nullptr)
    {
        std::cout << "something went wrong while creating the Transient Local Subscriber..." << std::endl;
        return false;
    }

#endif

    return true;
}

HelloWorldSubscriber::~HelloWorldSubscriber()
{
//    Domain::removeParticipant(mp_participant);
    for (auto reader:m_Reader)
    {
        m_subscriber->delete_datareader(reader.second.second);
        m_participant->delete_topic(reader.second.first);
    }
    m_Reader.clear();
    m_participant->delete_subscriber(m_subscriber);
    DomainParticipantFactory::get_instance()->delete_participant(m_participant);
}

/*void HelloWorldSubscriber::SubListener::onSubscriptionMatched(
        Subscriber* *//*sub*//*,
        MatchingInfo& info)
{
    if (info.status == MATCHED_MATCHING)
    {
        n_matched++;
//        std::cout << "Subscriber matched" << std::endl;
        Logger_Info("Subscriber matched");
    }
    else
    {
        n_matched--;
//        std::cout << "Subscriber unmatched" << std::endl;
        Logger_Info("Subscriber unmatched");
    }
}

void HelloWorldSubscriber::SubListener::onNewDataMessage(
        Subscriber* sub)
{
    if (sub->takeNextData((void*)&m_Hello, &m_info))
    {
        if (m_info.sampleKind == ALIVE)
        {
            this->n_samples++;
            // Print your structure data here.
            Logger_Info("[Message:{}, index:{}] RECEIVED", m_Hello.message(), m_Hello.index());
//            std::cout << "Message " << m_Hello.message() << " " << m_Hello.index() << " RECEIVED" << std::endl;
        }
    }

}*/

void HelloWorldSubscriber::run()
{
//    std::cout << "Subscriber running. Please press enter to stop the Subscriber" << std::endl;
    Logger_Info("Subscriber running. Please press enter to stop the Subscriber");
//    std::cin.ignore();
    SampleInfo sample_info;
    HelloWorld message;
//    std::string s;
//    while(std::cin >> s)
    while(true)
    {
        memset(&sample_info, 0, sizeof sample_info);
        memset(&message, 0, sizeof message);
        for (auto reader: m_Reader)
        {
            reader.second.second->read_next_sample(&message,&sample_info);
            if (sample_info.instance_state == ALIVE_INSTANCE_STATE)
            {
                Logger_Info("recv topic:{} Reader , index:{}", reader.first, message.index());
            }
        }
    }
}

/*
void HelloWorldSubscriber::run(
        uint32_t number)
{
//    std::cout << "Subscriber running until " << number << "samples have been received" << std::endl;
    Logger_Info("Subscriber running until {} samples have been received", number);
    while (number > this->m_listener.n_samples)
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(1));
    }
}*/

bool HelloWorldSubscriber::AddReaderChannel(std::string topic_name)
{
    if (nullptr == m_participant || nullptr == m_subscriber)
    {return false;}

    //Create Topic
    auto myTopic = m_participant->create_topic(topic_name, m_sampleType.get_type_name(), TOPIC_QOS_DEFAULT);
    if (myTopic == nullptr)
    {
        std::cout << "Something went wrong while creating the Topic..." << std::endl;
        return false;
    }

    DataReaderQos rqos1;
    rqos1.endpoint().history_memory_policy = DYNAMIC_RESERVE_MEMORY_MODE;
//    rqos1.history().kind = KEEP_ALL_HISTORY_QOS;
    rqos1.history().kind = KEEP_LAST_HISTORY_QOS;
    rqos1.durability().kind = TRANSIENT_LOCAL_DURABILITY_QOS; //如果 DataReader 在某个时间点之前未连接到 DataWriter，则该时间点之前写入的消息会被丢失
    rqos1.reliability().kind = RELIABLE_RELIABILITY_QOS;
    rqos1.history().depth =  10;
    rqos1.resource_limits().max_samples = 50;
//    rqos1.resource_limits().allocated_samples = 20;
    rqos1.resource_limits().max_instances = 1;
    rqos1.resource_limits().max_samples_per_instance = 10;

    auto myReader = m_subscriber->create_datareader(myTopic, rqos1);
//    auto myReader = m_subscriber->create_datareader(myTopic, DATAREADER_QOS_DEFAULT);

//    m_Reader[topic_name] = std::pair<Topic*,DataReader*>(myTopic, myReader);
    m_Reader.insert({topic_name,{myTopic, myReader}});

    return true;
}
